package org.example.mqactual.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.CountDownLatch2;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.TimeUnit;

/**
 * 发送异步消息
 */
public class AsyncProducer {

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, UnsupportedEncodingException {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr("192.168.254.129:9876");

        try {
            // 启动Producer实例
            producer.start();
            //发送单挑消息
            Message msg = new Message("test_topic", "hello rocketmq".getBytes());
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    //消息发送成功回调函数
                    System.out.printf("%s%n",sendResult);
                    System.out.println("发送成功");
                }

                //消息发送失败回调函数
                @Override
                public void onException(Throwable throwable) {
                    //发送失败，可以在这里做补偿，将消息存到数据库，定时重试

                }
            });
        }catch (Throwable e){
            e.printStackTrace();
            //发送失败，可以在这里做补偿，将消息存到数据库，定时重试
        }

        Thread.sleep(3000);
        //使用完毕后，关闭
        //springboot应用，消息发送的时候不会调用shutdown方法，而是等到spring容器停止
        producer.shutdown();

    }
}
